#YashanDB Kafka Connector

Debezium是专门为Kafka实现的CDC连接器，Debezium Connector YashanDB连接器同步全量快照数据，捕获并记录YashanDB数据库中发生的行级更改，包含连接器运行时添加的表。您可以配置YashanDB连接器，使其为特定的schema和表捕获更改事件，将更改事件同步到Kafka。

# 版本配套说明

# 前提条件

• 已安装Apache Zookeeper、Apache Kafka、 kafka Connect ，具体操作请查阅Apache Kafka官方文档 (opens new window)。

• 在kafka安装目录下创建plugins/debezium-connector-yashandb目录，并获取下列jar包并存放至该新建目录：

  可自行使用构建自动化工具（例如Maven）将项目构建打包，也可联系我们的技术支持获取。

  • debezium-connector-yashandb-2.4.2.jar

  • YStream-vxx.xx.xx.jar

  • yashandb-jdbc.jar

# 使用限制

数据类型不能为XMLTYPE、JSON以及自定义数据类型。

# 配置YashanDB

# 步骤1：配置Ystream内存池

增量数据依赖YashanDB的YStream实时获取YashanDB的已提交数据。

当您使用YashanDB作为源使用含增量同步的任务时，您需要在YashanDB中为YStream分配内存池：

ALTER SYSTEM SET STREAM_POOL_SIZE = streamPoolSize;

# 步骤2：开启附加日志

读取增量数据变更需要开启附加日志，可按需开启库级或表级附加日志。

Caution：

不开启附加日志或开启附加日志的对象不正确会导致数据丢失甚至任务失败。

当您需要监听库下全部对象时（包含新增对象），可开全库附加日志，方式如下：

ALTER DATABASE ADD SUPPLEMENTAL LOG TABLE TYPE (HEAP);
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA ( PRIMARY KEY) COLUMNS;

当您仅需要监听某些表时，可开启表级附加日志，方式如下：

ALTER TABLE tablename ADD SUPPLEMENTAL LOG DATA ( PRIMARY KEY ) COLUMNS;

Caution：

不开启附加日志或开启附加日志的对象不正确会导致数据丢失甚至任务失败。

# 步骤3：配置YStream服务

YStream服务的详细配置教程请查阅DBMS_YSTREAM_ADM高级包。

1. 调用DBMS_YSTREAM_ADM.CREATE函数创建YStream服务：

   EXEC DBMS_YSTREAM_ADM.CREATE('serverName', 'connect_user', start_scn)
   

   Copied!

   start_scn可通过查询select CURRENT_SCN from V$DATABASE获取。

2. 调用DBMS_YSTREAM_ADM.ADD_TABLES函数为YStream服务新增解析表名和模式（即connector需要捕获的表名和模式）。

3. 按需调用DBMS_YSTREAM_ADM.SET_PARAMETER函数为YStream服务设置相关参数。

4. 调用DBMS_YSTREAM_ADM.START函数启动YStream服务：

   DBMS_YSTREAM_ADM.START(server_name  IN VARCHAR(64));
   

   Copied!

# 步骤4：授予相关用户权限

连接用户需要相关权限，请授予如下权限保障任务能够正常运行：

GRANT CREATE SESSION TO username;
GRANT SELECT ON V_$DATABASE TO username;
GRANT SELECT ON V_$TRANSACTION TO username;
GRANT SELECT ON V_$YSTREAM_SERVER TO username;
GRANT FLASHBACK ANY TABLE TO username; 
GRANT SELECT ANY TABLE TO username;
GRANT YSTREAM_CAPTURE TO username;

# 部署debezium connector yashandb

1. 下载debezium相关依赖并存放至plugins/debezium-connector-yashandb目录，下载地址为https://repo1.maven.org/maven2/io/debezium/debezium-connector-oracle/2.4.2.Final/debezium-connector-oracle-2.4.2.Final-plugin.tar.gz (opens new window)。

2. 删除上述依赖包中的debezium-connector-oracle-2.4.2.Final.jar。

3. 将如下文件放入目录plugins/debezium-connector-yashandb中。

   • debezium-connector-yashandb-2.4.2.0.jar
   • YashanDB jdbc依赖
   • YashanDB YStream依赖

4. 将绝对路径/kafkahome/plugin/debezium-connector-yashandb添加到Kafka connect的配置文件的plugin.path中。

5. 重启Kafka Connect进程。

   bin/connect-distributed.sh config/connect-distributed.properties
   

   Copied!

6. 配置debezium YashanDB connector，如下：

   {
      "name": "yashandb-connector", 
      "config": {
         "connector.class" : "io.debezium.connector.yashandb.YashanDBConnector", 
         "database.hostname" : "<YashanDB_IP_ADDRESS>", 
         "database.port" : "1688", 
         "database.user" : "SYS", 
         "database.password" : "Cod-2022",  
         "database.dbname" : "TEST", 
         "topic.prefix" : "my_topic", 
         "database.url" : "jdbc:yasdb://<YashanDB_IP_ADDRESS>:1688/SYS", 
         "table.include.list" : "TEST.TAB01",
         "database.ystream.server.name" : "server1",  
         "lob.enabled" : "true",
         "schema.history.internal.kafka.bootstrap.servers" : "kafka:9092",
         "schema.history.internal.kafka.topic": "schema-changes.inventory" 
      }
   }
   

   Copied!

7. 启动任务。

# 连接器参数

本文仅列示部分关键参数，更多详情请参考Debezium Connector for Oracle官方文档 (opens new window)。

# 数据类型映射

当Debezium Oracle连接器检测到表行的值发生更改时，会发出表示该更改的更改事件。每个更改事件记录的结构与原始表的结构相同，事件记录包含每个列值的字段。表列的数据类型决定了连接器如何在更改事件字段中表示列的值，如以下各节中的表所示。

对于表中的每一列，Debezium将源数据类型映射到literal文本类型。

文本类型

使用以下Kafka Connect架构类型之一描述如何按字面表示值：INT8、INT16、INT32、INT64、FLOAT32、FLOAT64、BOOLEAN、STRING、BYTES、ARRAY、MAP、STRUCT。

# 常见问题

# Q1. 报错“YashanDB does not yet have the YStream server ‘serverxx’ or check option 'database.ystream.server.name' if the parameters are filled in correctly. Please create and configure the YStream server, refer to the link 'xxx'.”该如何处理？

该报错表示database.ystream.server.name参数值对应的YStream服务不存在，请先调用DBMS_YSTREAM_ADM高级包对应函数创建YStream服务并完成相关配置。

# Q2. 报错：YashanDB YStream server status is xxx. Please execute 'DBMS_YSTREAM_ADM.START( server_name IN VARCHAR(64) );' start YStream server。

该报错表示database.ystream.server.name参数值对应的YStream服务处于非运行状态或者非启动状态，请先在数据库中调用DBMS_YSTREAM_ADM.START函数启动该YStream服务：

exec DBMS_YSTREAM_ADM.START('YStream服务名');

# Q3. Decimal数值同步到Kafka后，为什么序列化出来数据出错？

debezium会将负scale的Decimal进行特殊处理，建议使用参数decimal.handling.mode=string来规避。